// TODO get all types from zookeeper, then check rpc parameter [type]
// 


var zookeeper = require('node-zookeeper-client');
var amqp = require('amqplib/callback_api');
var util = require('util');
var EventEmitter = require('events').EventEmitter;

// zk
var __zkNode = '/myrpc/server/currid';
var __zkConfig = 'localhost:2181';
var __zkBaseId = '1';
// mq
var __rabbitmqConfig = 'amqp://localhost';
var __qRequest = 'request_to_{type}';
var __qReply = 'reply_to_{appId}';


// all events
// error, err
// zkDone
// mqDone
function App(type, handler, errHandler) {
	if (this instanceof App) {
		console.log('init');
		this.type = type;
		this.handler = handler;
		this.errHandler = errHandler;
		this.rpcCallback = {};
		this.msgId = 0;
		// 
		this.prepareZk();
		return this;
	} else {
		return new App(type, handler, errHandler);
	}
}

util.inherits(App, EventEmitter);


App.prototype.prepareZk = function() {
	console.log('prepareZk');
	var self = this;
	var zk = zookeeper.createClient(__zkConfig);
	self.__zk = zk;
	zk.connect();

	function setMyId(id) {
		zk.setData(__zkNode, new Buffer(id + ''), -1, function(err, stat) {
			if (err) return self.onError(err);
			self.appId = id;
			// self.emit('zkDone');
			self.prepareMq();
		});
	}

	function getId() {
		zk.getData(__zkNode, null, function(err, data, stat) {
			// console.log('getData node %s', __zkNode);
			if (err) return self.onError(err);
			var myId = Number(data) + 1;
			self.appId = myId;
			setMyId(myId);
		});
	}

	function firstSetId() {
		zk.mkdirp(__zkNode, new Buffer(__zkBaseId), function(err, path) {
			// console.log('create node %s', __zkNode);
			if (err) return self.onError(err);
			self.appId = __zkBaseId;
			// self.emit('zkDone');
			self.prepareMq();
		});
	}


	zk.once('connected', function() {
		zk.exists(__zkNode, function(err, stat) {
			if (err) return self.onError(err);
			if (stat) {
				// console.log('exists node %s', __zkNode);
				getId();
			} else {
				// console.log('not exists node %s', __zkNode);
				firstSetId();
			}
		});
	});
};

App.prototype.prepareMq = function() {
	console.log('prepareMq');
	var self = this;
	amqp.connect(__rabbitmqConfig, function(err, conn) {
		if (err) return self.onError(err);

		conn.createChannel(function(err, ch) {
			if (err) return self.onError(err);
			self.__mqCh = ch;
			self.__mqConn = conn;
			//
			prepareMqReplyChannel();
		});
	});

	function prepareMqReplyChannel() {
		// self.__mqConn.createChannel(function(err, ch) {
		ch = self.__mqCh; // reply channel prefetch not set to 1
		var qReply = __qReply.replace('{appId}', self.appId);
		ch.assertQueue(qReply, {
			exclusive: true
		}, function(err, q) {
			console.log('create queue %s', qReply);
			ch.consume(qReply, function(msg) {
				console.log('receive reply at ', new Date().getTime());
				ch.ack(msg); // ack
				// console.log('receive reply: %s', msg.content);
				var msgId = msg.properties.correlationId;
				var cb = self.rpcCallback[msgId];
				if (!cb) {
					console.error('no callback with msgId = %s', msgId);
				} else {
					cb(err, msg.content);
				}
			});
			prepareMqRequestChannel();
		});
		// });
	}

	function prepareMqRequestChannel() {
		var ch = self.__mqCh;
		var qRequst = __qRequest.replace('{type}', self.type);
		console.log('create queue %s', qRequst);
		ch.assertQueue(qRequst, {
			durable: true
		});
		ch.prefetch(1); // TODO fetch reply should not set prefetch =1 ,but fetch request should set to 1 
		ch.consume(qRequst, function(msg) {
			console.log('receive request at ', new Date().getTime());
			// console.log('receive request:%s', msg.content);
			ch.ack(msg);
			self.handler(msg, function(result) {
				var replyTo = msg.properties.replyTo;
				var replyQid = __qReply.replace('{appId}', replyTo);
				// console.log('handle done, replyQid=%s, result=%s', replyQid, result);
				console.log('send reply at ', new Date().getTime());
				ch.sendToQueue(replyQid, new Buffer(result), {
					correlationId: msg.properties.correlationId
				});
			});
		}, {
			noAck: false
		});
		self.prepared = true;
		self.emit('mqDone');
	}
};


/**
 * cb( err, function(result){...})
 */
App.prototype.rpc = function(type, arg, cb) {
	// console.log('rpc', type, arg);
	var self = this;
	if (!self.prepared) return self.onError(new Error('App not prepared'));
	var ch = self.__mqCh;
	// send to mq
	var qSend = __qRequest.replace('{type}', type);
	ch.assertQueue(qSend, {
		durable: true
	}, function(err, q) {
		var msgId = (self.msgId++) + '';
		console.log('send request at ', new Date().getTime());
		ch.sendToQueue(qSend, new Buffer(arg), {
			correlationId: msgId,
			replyTo: self.appId + '',
			persisitent: true,
		});
		// put callback to list
		self.rpcCallback[msgId] = cb;
	});
};


/**
 * cb( err, function(result){...})
 */
App.prototype.rpcBenchmark = function(type, arg, time, cb) {
	// console.log('rpc', type, arg);
	var self = this;
	if (!self.prepared) return self.onError(new Error('App not prepared'));
	var ch = self.__mqCh;
	// send to mq
	var qSend = __qRequest.replace('{type}', type);
	ch.assertQueue(qSend, {
		durable: true
	}, function(err, q) {
		var msgId = (self.msgId++) + '';
		console.log('send request at ', new Date().getTime());
		ch.sendToQueue(qSend, new Buffer(arg), {
			correlationId: msgId,
			replyTo: self.appId + '',
			persisitent: true,
		});
		// put callback to list
		self.rpcCallback[msgId] = cb;
	});
};

App.prototype.onError = function(err) {
	if (this.errHandler) {
		return this.errHandler(err);
	}
	throw new Error('errHandler not set, uncaught error', err);
};


module.exports = App;